package contegor.scire.structure;

import static contegor.scire.structure.Constants.TO_DATE;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.FilenameFilter;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.net.URI;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.ExecutionException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.CacheLoader.InvalidCacheLoadException;
import com.google.common.cache.LoadingCache;

import contegor.scire.core.Action;
import contegor.scire.core.Concept;
import contegor.scire.core.ConceptImpl;
import contegor.scire.core.Connectable;
import contegor.scire.core.ConnectableImpl;
import contegor.scire.core.Container;
import contegor.scire.core.ContainerImpl;
import contegor.scire.core.DuplicateElementException;
import contegor.scire.core.Id;
import contegor.scire.core.IdFragment;
import contegor.scire.core.Identifiable;
import contegor.scire.core.Link;
import contegor.scire.core.Link.Flow;
import contegor.scire.core.LinkImpl;
import contegor.scire.core.MissingElementException;
import contegor.scire.core.NonDeletableElementException;
import contegor.scire.core.Primitive;
import contegor.scire.core.PrimitiveImpl;
import contegor.scire.core.ReturnPoint;
import contegor.scire.core.StorageResultsReturnPoint;
import contegor.scire.core.StringIdFragment;
import contegor.scire.core.TimestampIdFragment;
import contegor.scire.core.TrackExternalLinks;
import contegor.scire.core.Transaction;
import contegor.scire.core.TransactionException;
import contegor.scire.core.TransactionId;
import contegor.scire.core.WrongStateException;

/**
 * {@link Container} who's {@link Concept}s are the files present in the directory given as root and,
 * recursively, those in its sub-directories.
 * 
 * It creates a hidden sub-directory named '.scire' where {@link Link}'s info is persisted in two
 * ways:
 * 
 * {@link Id} + '.reverse' named files. Here a reverse index of stored {@link Link}s. For each target
 * {@link Id} of any {@link Link}, a file is created who's name is the full target {@link Id} and its
 * content is a serialized {@link Set} of references to all {@link Link}s that aim that {@link Id}.
 * This {@link Link}s are referenced through its successive {@link IdFragment} and {@link Id}s that serve
 * as hops to identify it as with {@link ContainerImpl#findConnectable(IdFragment, Id[])}.
 * 
 * {@link Id} + '.links' named files. Here are stored the {@link Concept} specific {@link Link}s. The
 * {@link Concept}'s {@link Id} gives name to the file. The {@link Link}s are serialized to this
 * file as an array.
 * 
 * Data is saved to this files in commit/rollback operations.
 * 
 * This structures are only for {@link Link}s specific to {@link Concept}s inside <code>this</code>. Beside
 * this ones, there are {@link Link}s that are common to all contained {@link Concept}s. For example, each
 * file will have always a 'last modification time', and this will be expressed with the appropriate {@link Link}
 * from its {@link Concept}. This 'fixed' {@link Link}s are not stored in those files and are managed
 * in an alternative and common way to all {@link Concept}s.
 * 
 * @version $Revision$
 * @author  Héctor Cirbián Pousa ($Author$)
 */
public class FilesContainer extends ContainerImpl {
  private static final long serialVersionUID = 3078309504815095597L;
  private static final Logger logger = LoggerFactory.getLogger(FilesContainer.class);

  // TODO: Take this of when no more needed.
  protected static final Id CREATION_TIME = new Id(Id.RAW_ROOT, new StringIdFragment("CREATION_TIME"));

  public static final String REFERENCE_FILE_SUFIX = ".reference";
  public static final String LINKS_FILE_SUFIX = ".links";

  /** Wraps a {@link Cache} so it can be used as a {@link ConcurrentMap} but using also {@link Cache}s load implementation when calling {@link Map#get(Object)}*/
  protected static class ConcurrentMapCacheWrapper<K, V> implements ConcurrentMap<K, V> {

    private LoadingCache<K, V>  cache;
    private ConcurrentMap<K, V> map;

    public ConcurrentMapCacheWrapper(LoadingCache<K, V> cache) {
      this.cache = cache;
      this.map = cache.asMap();
    }

    @Override
    public V putIfAbsent(K key, V value) {
      return this.map.putIfAbsent(key, value);
    }

    @Override
    public boolean remove(Object key, Object value) {
      return this.map.remove(key, value);
    }

    @Override
    public V replace(K key, V value) {
      return this.map.replace(key, value);
    }

    @Override
    public boolean replace(K key, V oldValue, V newValue) {
      return this.map.replace(key,oldValue,newValue);
    }

    @Override
    public void clear() {
      this.map.clear();
    }

    @Override
    public boolean containsKey(Object key) {
      return this.map.containsKey(key);
    }

    @Override
    public boolean containsValue(Object key) {
      return this.map.containsValue(key);
    }

    @Override
    public Set<java.util.Map.Entry<K, V>> entrySet() {
      return this.map.entrySet();
    }

    @Override
    @SuppressWarnings("unchecked")
    public V get(Object key) {
      try {
        return this.cache.get((K)key);
      }
      catch (InvalidCacheLoadException ex) {
        logger.info(ex.getMessage(), ex);
        return null;
      }
      catch (ExecutionException ex) {
        logger.error(ex.getMessage(), ex);
        return null;
      }
    }

    @Override
    public boolean isEmpty() {
      return this.map.isEmpty();
    }

    @Override
    public Set<K> keySet() {
      return this.map.keySet();
    }

    @Override
    public V put(K key, V value) {
      return this.map.put(key, value);
    }

    @Override
    public void putAll(Map<? extends K, ? extends V> map) {
      this.map.putAll(map);
    }

    @Override
    public V remove(Object key) {
      return this.map.remove(key);
    }

    @Override
    public int size() {
      return this.map.size();
    }

    @Override
    public Collection<V> values() {
      return this.map.values();
    }
    
  }

  protected static class LinkReference implements Serializable, Comparable<LinkReference>{
    private static final long serialVersionUID = 4820650738429090974L;

    private IdFragment startPoint;
    private Id[]       nextPoints;

    public LinkReference(IdFragment startPoint, Id[] nextPoints) {
      super();
      this.startPoint = startPoint;
      this.nextPoints = nextPoints;
    }

    public IdFragment getStartPoint() {
      return startPoint;
    }

    public void setStartPoint(IdFragment startPoint) {
      this.startPoint = startPoint;
    }

    public Id[] getNextPoints() {
      return nextPoints;
    }

    public void setNextPoints(Id[] nextPoints) {
      this.nextPoints = nextPoints;
    }

    @Override
    public int hashCode() {
      final int prime = 31;
      int result = 1;
      result = prime * result + Arrays.hashCode(nextPoints);
      result = prime * result + ((startPoint == null) ? 0 : startPoint.hashCode());
      return result;
    }

    @Override
    public boolean equals(Object obj) {
      if (this == obj)
        return true;
      if (obj == null)
        return false;
      if (getClass() != obj.getClass())
        return false;
      LinkReference other = (LinkReference) obj;
      if (!Arrays.equals(nextPoints, other.nextPoints))
        return false;
      if (startPoint == null)
        return other.startPoint == null;
      return startPoint.equals(other.startPoint);
    }

    @Override
    public int compareTo(LinkReference lhs) {
      int aux = this.startPoint.compareTo(lhs.startPoint);
      if (aux != 0)
        return aux;
      for (int i=0; ((i<this.nextPoints.length) && (i<lhs.nextPoints.length)); i++) {
        aux = this.nextPoints[i].compareTo(lhs.nextPoints[i]);
        if (aux != 0)
          return aux;
      }
      return this.nextPoints.length - lhs.nextPoints.length;
    }
  }

  /** Used to represent the fixed {@link Link}s every {@link Concept} in this {@link Container} should have.*/
  protected static class FixedLink extends LinkImpl {
    private static final long serialVersionUID = -1384471765259166547L;

    protected FixedLink(Connectable source, Flow flow, Id target) {
      super(0, null, flow);
      this.setSource(source);
      this.setTarget(target);
    }

    @Override
    protected boolean connect(LinkImpl link) {
      return super.connect(link);
    }

    protected Id[] calcHops() {
      List<Id> result = new LinkedList<Id>();
      Link link = this;
      do {
        result.add(0, link.getTarget());
        Connectable aux = link.getSource();
        if (aux != null) {
          link = (Link)aux;
          continue;
        }
      }
      while (false);
      return result.toArray(new Id[result.size()]);
    }
  }

  /**
   * When using this {@link TransactionId}, commit will not persist any data changed by it.
   * Used to set, in internal transient structures, as caches, data that is already persisted.  
   */
  protected static class NotPersistentTransactionId extends TransactionId {
    private static final long serialVersionUID = 7053899604105395525L;
    public static final String NOT_PERSISTENT = "NOT_PERSISTENT";
    public NotPersistentTransactionId() {
      super(NOT_PERSISTENT,false);
    }
  }

  protected List<FixedLink> rootFixedLinks;
  protected Map<Id, List<FixedLink>> reverseFixedLinkReference;

  protected File root;
  protected File data;
  protected ConcurrentMap<Id, Set<LinkReference>> linkReferencesCache; 

  public FilesContainer(long creationTime, TransactionId transactionId, IdFragment idFragment, File root) throws WrongStateException, IOException {
    super(creationTime, transactionId, idFragment);
    if (!root.isDirectory())
      throw new WrongStateException("The given path is not a directory: " + root.getAbsolutePath());
    this.root = root.getCanonicalFile();
    data = new File(this.root, ".scire");
    if (data.exists()) {
      if (!data.isDirectory())
        throw new WrongStateException(root.getAbsolutePath() + ".scire exists and is not a directroy.");
    }
    else {
      if (!data.mkdir())
        throw new WrongStateException("It was not possible to create: " + data.getAbsolutePath() + ".scire");
    }
  }

  public FilesContainer(long creationTime, TransactionId transactionId, String idFragment, String root) throws WrongStateException, IOException {
    this(creationTime, transactionId, new StringIdFragment(idFragment), new File(root));
  }

  public FilesContainer(long creationTime, TransactionId transactionId, String idFragment, URI root) throws WrongStateException, IOException {
    this(creationTime, transactionId, new StringIdFragment(idFragment), new File(root));
  }

  public FilesContainer(String idFragment, File root) throws WrongStateException, IOException {
    this(System.currentTimeMillis(), null, new StringIdFragment(idFragment), root);
  }

  public FilesContainer(String idFragment, String root) throws WrongStateException, IOException {
    this(System.currentTimeMillis(), null, new StringIdFragment(idFragment), new File(root));
  }

  protected void init() {
    setUpFixedLinks();
    this.externalReferences = new ConcurrentSkipListMap<LocalIdAndContainerId, AtomicIntegerWithEquals>();
    this.transactionPendingConnectables = new ConcurrentSkipListMap<TransactionId, Set<ConnectableImpl>>();
    this.identifiables = new ConcurrentMapCacheWrapper<IdFragment, Identifiable>(CacheBuilder.newBuilder().softValues().build(
      new CacheLoader<IdFragment, Identifiable>() {
        public Identifiable load(IdFragment id) {
          return readConceptFromFile(id);
        }
      }));
    this.reverseLinkReferences = new ConcurrentMapCacheWrapper<Id, Set<LinkImpl>>(CacheBuilder.newBuilder().softValues().build(
      new CacheLoader<Id, Set<LinkImpl>>() {
        public Set<LinkImpl> load(Id aimedId) {
          return retrieveReverseLinkReferences(aimedId);
        }
      }));
    this.linkReferencesCache = new ConcurrentMapCacheWrapper<Id, Set<LinkReference>>(CacheBuilder.newBuilder().softValues().build(
      new CacheLoader<Id, Set<LinkReference>>() {
        public Set<LinkReference> load(Id aimedId) {
          return retreiveLinkReferences(aimedId);
        }
      }));
  }

  protected void setUpFixedLinks() {
    // Set as an "actor --- action ---> object" configuration (the actor is the date, and the object the file).
    // But the link structure is set in the 'object' realm, with an 'upstream' link. This is reverse to
    // the usual way of setting this, but the only way when the 'actor' is raw data, not even a primitive.
    FixedLink date = new FixedLink(null, Flow.UPSTREAM, CREATION_TIME);
    FixedLink to_date = new FixedLink(date, Flow.NONE, TO_DATE);
    date.connect(to_date);

    rootFixedLinks = new ArrayList<FixedLink>(1);
    rootFixedLinks.add(date);
    reverseFixedLinkReference = new TreeMap<Id, List<FixedLink>>();
    reverseFixedLinkReference.put(TO_DATE, Arrays.asList( new FixedLink[] {to_date}));
  }

  private File fileForLinksReferences(Id id) {
    return new File(this.data, id.toString().replace(File.separatorChar, '_') + REFERENCE_FILE_SUFIX);    
  }

  private File fileForConceptLinks(Id id) {
    return new File(this.data, id.toString().replace(File.separatorChar, '_') + LINKS_FILE_SUFIX);
  }

  protected Concept readConceptFromFile(IdFragment id) {
    try {
      File file = new File(this.root, id.toString());
      if (!file.isFile())
        return null;
      TransactionId transactionId = new NotPersistentTransactionId();
      ConceptImpl concept = (ConceptImpl)super.createPrimitive(file.lastModified(), transactionId, id, new URI("file", file.getAbsolutePath(), null));
      // Fixed links.
      createFixedLinks(concept, file, transactionId);
      // Saved links.
      File linksFile = fileForConceptLinks(concept.getId());
      if (linksFile.isFile()) {
        try (ObjectInputStream ois = new ObjectInputStream(new FileInputStream(linksFile))) {
          for (LinkImpl link: (LinkImpl[])ois.readObject()) {
            this.connect((ConnectableImpl)concept, link);
            // This is causes recursive loads and it's not needed:
            // this.addLinkToInternalStructures(transactionId, link);
          }
        }
      }
      this.commit(transactionId);
      return concept;
    }
    catch (Exception ex) {
      logger.error(ex.getMessage(), ex);
      return null;
    }
  }

  // This is causes recursive loads and it's not needed.
  //  private void addLinkToInternalStructures(TransactionId transactionId, LinkImpl link) {
  //    addToTransaction(transactionId, link);
  //    addReverseLinkReference(link, link.getTarget());
  //    for (Link child: link.getLinks().values())
  //      addLinkToInternalStructures(transactionId, (LinkImpl)child);
  //  }

  /**
   * Creates the fixed {@link Link}s for a {@link Concept} based on its {@link File}. Implementations
   * of this method should extract from the {@link Concept}'s {@link File} the data needed by the 
   * extended {@link FilesContainer}'s {@link FixedLink}s, and then call with it the
   * {@link #createFixedLinks(ConceptImpl, long, TransactionId, Map)} method. That extracted data will
   * be passed as {@link IdFragment}s. Te get this thata use only source file content or the file's name
   * or the concept's content, but never file's path.
   * 
   * @param concept {@link ConceptImpl} to which create the fixed {@link Link}s.
   * @param file {@link File} from which extract the data the fixed {@link Link}s will need.
   * @param transactionId {@link TransactionId} that represents the transaction under this modification is been done.
   * @throws WrongStateException 
   * @throws TransactionException 
   * @throws DuplicateElementException 
   */
  protected void createFixedLinks(ConceptImpl concept, File file, TransactionId transactionId) throws DuplicateElementException, TransactionException, WrongStateException {
    Map<Id,IdFragment> rawData = new TreeMap<Id, IdFragment>();
    rawData.put(CREATION_TIME, new TimestampIdFragment(file.lastModified()));
    createFixedLinks(concept, file.lastModified(), transactionId, rawData);    
  }

  /**
   * Creates the 'fixed' {@link Link}s for a concept, using the ones of {@link #setUpFixedLinks()} as model.
   * The target {@link Id}s from {@link #setUpFixedLinks()} set to {@link Id#RAW_ROOT} are changed to
   * {@link Id#RAW_ROOT} + the {@link IdFragment} passed in rawData parameter. This substitution is done
   * in recursive order.
   * The {@link Link}s created this way are not modifiable, and no {@link Link}s can be attached to them.
   * 
   * @param concept {@link ConceptImpl} for which the {@link Link}s will be created.
   * @param creationTime long with the creation time that will be given to the {@link Link}s.
   * @param trId {@link TransactionId} which under the {@link Link}s will be created.
   * @param rawData {@link IdFragment} with which substitute {@link Id#RAW_ROOT} targets of fixed {@link Link} models.
   */
  protected final void createFixedLinks(ConceptImpl concept, long creationTime, TransactionId trId, Map<Id,IdFragment> rawData) throws DuplicateElementException, TransactionException, WrongStateException {
    for (FixedLink model: rootFixedLinks)
      createFixedLink(creationTime, trId, concept, model, rawData);
  }

  private void createFixedLink(long creationTime, TransactionId trId, ConnectableImpl source, FixedLink model, Map<Id,IdFragment> rawData) throws DuplicateElementException, TransactionException, WrongStateException {
    Id target;
    if (Id.RAW_ROOT.isAncestorOf(model.getTarget())) {
      IdFragment data = rawData.get(model.getTarget());
      if (data != null)
        target = new Id(model.getTarget().getParentId(), data);
      else
        return; // If there is no data for this target, is not created.
        // FIXME: The result are not valid fixed links reverse references ....
        // Perhaps is better to create an 'unknown' target.
        // This unknown target should be comparable with other IdFragments.
    }
    else
      target = model.getTarget();
    LinkImpl link = this.onlyCreateLinkImpl(creationTime, trId, model.getFlow(), source, target);
    link.setStatus(Status.FIXED);
    Map<Id,Link> childs = model.getLinks();
    for (Link child: childs.values())
      createFixedLink(creationTime, trId, link, (FixedLink)child, rawData);
  }

  @Override
  protected void addReverseLinkReference(LinkImpl link, Id target) {
    if (link.getStatus() != Status.FIXED)
      super.addReverseLinkReference(link, target);
  }
  
  protected Set<LinkImpl> retrieveReverseLinkReferences(Id aimedId) {
    try {
      Set<LinkImpl> result = new ConcurrentSkipListSet<LinkImpl>();
      Set<LinkReference> reverseLinkReferences = this.linkReferencesCache.get(aimedId);
      if (reverseLinkReferences != null) {
        for (LinkReference reverseLinkReference: reverseLinkReferences) {
          // This will call indirectly to this.identifiables.get(link_root_concept) and this will make it and all its links to be loaded if they are not cached.
          result.add((LinkImpl)this.findConnectable(reverseLinkReference.getStartPoint(), reverseLinkReference.getNextPoints()));
        }
      }
      // Code to retrieve fixed links who's target is aimedId.
      // If the aimedId is target of any Fixed Link, then are retrieved for all concepts using also findConnectable.
      List<FixedLink> fixedLinks = this.reverseFixedLinkReference.get(aimedId);
      if (fixedLinks != null) {
        Set<IdFragment> localIds = this.getLocalIds();
        for (FixedLink fixed: fixedLinks) {
          Id[] hops = fixed.calcHops();
          for (IdFragment localId: localIds) {
            // This will call indirectly to this.identifiables.get(link_root_concept) and this will make it and all its links to be loaded if they are not cached.
            result.add((LinkImpl)this.findConnectable(localId, hops));            
          }
        }
      }
      return result;
    }
    catch (Exception ex) {
      logger.error(ex.getMessage(), ex);
      return null;
    }
  }

  @SuppressWarnings("unchecked")
  protected Set<LinkReference> retreiveLinkReferences(Id aimedId) {
    try {
      File file = fileForLinksReferences(aimedId);
      if (file.isFile()) {
        try (ObjectInputStream ois = new ObjectInputStream(new FileInputStream(file))) {
          return (Set<LinkReference>)ois.readObject();
        }
      }
      return new TreeSet<LinkReference>();
    }
    catch (Exception ex) {
      logger.error(ex.getMessage(), ex);
      return null;
    }
  }

  protected void persistLinksReferences(Id target, Set<Link> links) throws WrongStateException, FileNotFoundException, IOException {
    Set<LinkReference> reverseLinkReferences = linkReferencesCache.get(target);
    for (Link link:links) {
      if (!target.equals(link.getTarget()))
        throw new WrongStateException(link + " 's target is not the expected");
      List<Id> linksTargets = new LinkedList<Id>();
      linksTargets.add(0, link.getTarget());
      Link aux = link;
      while (aux.getSource() instanceof Link) {
        aux = (Link)aux.getSource();
        linksTargets.add(0, aux.getTarget());
      }
      LinkReference reference = new LinkReference(((Identifiable)aux.getSource()).getLocalIdFragment(), linksTargets.toArray(new Id[linksTargets.size()]));
      if (((ConnectableImpl)link).getStatus() == Status.ERASED)
        reverseLinkReferences.remove(reference);
      else if (!reverseLinkReferences.contains(reference))
        reverseLinkReferences.add(reference);
    }
    synchronized (reverseLinkReferences) {
      try (ObjectOutputStream oos = new ObjectOutputStream(new FileOutputStream(fileForLinksReferences(target)))) {
        oos.writeObject(reverseLinkReferences);
      }
    }
  }

  @Override
  public boolean contains(IdFragment id) {
    return this.getConcept(id) != null;
  }

  @Override
  public Set<IdFragment> getLocalIds() {
    Set<IdFragment> localIds = new TreeSet<IdFragment>();
    retrieveLocalIds(root.getAbsolutePath().length()+1, root, localIds);
    return localIds;
  }

  private void retrieveLocalIds(int rootLength, File subdir, Set<IdFragment> localIds) {
    for (File file: subdir.listFiles()) {
      if (!file.isHidden()) {
        if (file.isDirectory())
          retrieveLocalIds(rootLength, file, localIds);
        else if (file.isFile()) {
          String path = file.getAbsolutePath();
          localIds.add(new StringIdFragment(path.substring(rootLength, path.length())));
        }
      }
    }
  }

  @Override
  public List<Concept> getConcepts() {
    Set<IdFragment> localIds = getLocalIds();
    List<Concept> concepts = new ArrayList<Concept>(localIds.size());
    for (IdFragment localId: localIds)
      concepts.add(this.getConcept(localId));
    return concepts;
  }

  @SuppressWarnings("unchecked")
  @Override
  public Map<Id, Integer> getNumberOfLinksTo(Id root) {
    Map<Id, Integer> result = new TreeMap<Id, Integer>();
    try {
      final String rootId = root.toString();
      Integer localIdsCount = null; 
      File[] references = data.listFiles(new FilenameFilter() {
        @Override
        public boolean accept(File dir, String name) {
          return name.startsWith(rootId) && name.endsWith(REFERENCE_FILE_SUFIX);
        }
      });
      if (references != null) {
        for (File refFile: references) {
          if (refFile.isFile()) {
            try (ObjectInputStream ois = new ObjectInputStream(new FileInputStream(refFile))) {
              Set<LinkReference> linkReferences = (Set<LinkReference>)ois.readObject();
              Id[] nextPoints = linkReferences.iterator().next().getNextPoints();
              Id target = nextPoints[nextPoints.length-1];
              Integer num = result.get(target);
              if (num == null)
                result.put(target, linkReferences.size());
              else
                result.put(target, num + linkReferences.size());
            }
          }
        }
      }
      for (Id target: this.reverseFixedLinkReference.keySet()) {
        if (root.isAncestorOf(target)) {
          if (localIdsCount == null)
            localIdsCount = this.size();
          int fixedNum = localIdsCount;
          List<FixedLink> fixedLinks = this.reverseFixedLinkReference.get(target);
          if (fixedLinks != null) {
            fixedNum *= fixedLinks.size();
          }
          Integer num = result.get(target);
          if (num == null)
            result.put(target, fixedNum);
          else
            result.put(target, num + fixedNum);
        }        
      }
    }
    catch (Exception ex) {
      logger.error(ex.getMessage(), ex);
    }
    return result;
  }

  private int countLocalIds(File subdir) {
    int count = 0;
    for (File file: subdir.listFiles()) {
      if (!file.isHidden()) {
        if (file.isDirectory())
          count += countLocalIds(file);
        else if (file.isFile())
          count++;
      }
    }
    return count;
  }

  @Override
  public int size() {
    return countLocalIds(root);
  }

  @Override
  public Concept createConcept(long creationTime, TransactionId transactionId, IdFragment idf) throws DuplicateElementException {
    throw new UnsupportedOperationException();
  }

  @Override
  public Action createAction(long creationTime, TransactionId transactionId, IdFragment idf, String plugin, String implementation) throws DuplicateElementException {
    throw new UnsupportedOperationException();
  }

  /*
   * In value we have an URI with the source file.
   * The idf denotes where that file will be copied.
   */
  @Override
  public Primitive createPrimitive(long creationTime, TransactionId transactionId, IdFragment idf, Object value) throws DuplicateElementException {
    try {
      if ((value != null) && (value instanceof URI)) {
        URI uri = ((URI)value);
        if (uri.getScheme().equals("file")) {
          // Checks.
          File src = new File(uri);
          if (!src.exists())
              throw new IllegalArgumentException("Trying to create " + idf + " in " + this.getId() + " giving a non existent source for it:" + uri);
          File dst = new File(this.root, idf.toString());
          File dstParent = dst.getParentFile();
          if (!dstParent.isDirectory() && !dstParent.mkdirs())
            throw new IllegalStateException("It was not possible to create directories for " + idf);
          if (!dst.createNewFile())
            throw new DuplicateElementException("File " + idf + " already in " + this.getId());
          PrimitiveImpl result = (PrimitiveImpl)super.createPrimitive(creationTime, transactionId, idf, uri);
          if (transactionId != null)
            addToTransaction(transactionId, result);
          createFixedLinks(result, src, transactionId);
          // It's a new Primitive, so build TrackExternalLinks tasks for fixed links.
          StorageResultsReturnPoint<Void> rp = new StorageResultsReturnPoint<Void>();
          sendTrackExteranlLinks(result, new Transaction(true), rp);
          // FIXME: Check possible ReturnPoint exceptions. Problem: concurrency ...
          return result;
        }
        else
          throw new IllegalArgumentException("Usupported URI: " + uri);
      }
      else
        return (Primitive)readConceptFromFile(idf);
    }
    catch (Exception ex) {
      throw new IllegalStateException(ex);
    }
  }

  private void sendTrackExteranlLinks(Connectable connectable, Transaction transaction, ReturnPoint<Void> rp) {
    for (Link link: connectable.getLinks().values()) {
      sendTrackExteranlLinks(link, transaction, rp);
      Id targetId = link.getTarget();
      if (!Id.RAW_ROOT.isAncestorOf(targetId))
        this.route(new TrackExternalLinks(transaction.nextTaskId(), targetId.getParentId(), targetId.getLastIdFragment(), this.getId()), rp);
    }
  }

  @Override
  public void incorporateContainer(Container container) throws DuplicateElementException, WrongStateException {
    throw new UnsupportedOperationException();
  }

//  @Override
//  public Link createLink(long creationTime, TransactionId transactionId, Flow flow, Connectable source, Id target) throws WrongStateException, DuplicateElementException, TransactionException {
//    return super.createLink(creationTime, transactionId, flow, source, target);
//  }

//  @Override
//  public Link createLink(long creationTime, TransactionId transactionId, Flow flow, Id target, IdFragment sourceStartPoint, Id... sourceNextPoints) throws MissingElementException, DuplicateElementException, TransactionException {
//    return super.createLink(creationTime, transactionId, flow, target, sourceStartPoint, sourceNextPoints);
//  }

  @Override
  public Primitive setPrimitiveValue(long modificationTime, TransactionId transactionId, IdFragment idf, Object value) throws MissingElementException, WrongStateException, TransactionException {
    throw new UnsupportedOperationException();
  }

  @Override
  public void deleteElement(TransactionId transactionId, Set<Id> insideLinksErased, IdFragment startPoint, Id... nextPoints) throws MissingElementException, NonDeletableElementException, TransactionException {
    ConnectableImpl connectable = findConnectable(startPoint, nextPoints);
    if (connectable == null)
      throw new MissingElementException("The connectable does not belong to this container: " + startPoint + " " + nextPoints);
    if (connectable instanceof Identifiable) // FIXME: This should be possible in the future.
      throw new NonDeletableElementException("Trying to delete an Identifiable. In this Container this is not possible.");
    if ((connectable instanceof LinkImpl) && ((LinkImpl)connectable).getStatus() == Status.FIXED)
      throw new NonDeletableElementException("Trying to delete a Fixed Link: " + connectable);
    markAsDeleted(transactionId, connectable);
  }

  @SuppressWarnings("incomplete-switch")
  @Override
  public List<Id> commit(TransactionId transactionId) throws TransactionException, MissingElementException {
    Set<ConnectableImpl> connectables = this.transactionPendingConnectables.get(transactionId);
    if (connectables == null)
      return null; // This transaction changed nothing.
    // First we collect created/modified/deleted elements.
    Map<Id,Set<Link>> linksToPersist = new TreeMap<Id, Set<Link>>();
    Set<Concept> conceptsToPersist = new HashSet<Concept>(connectables.size());
    if (!(transactionId instanceof NotPersistentTransactionId)) {
      for (ConnectableImpl connectable: connectables) {
        if (connectable instanceof Link) {
          Link link = (Link)connectable;
          switch (connectable.getStatus()) {
            case ERASED:
            case CREATED: { // In case a Link has been created or erased, its references must be updated.
              Set<Link> links = linksToPersist.get(link.getTarget());
              if (links == null)
                linksToPersist.put(link.getTarget(), links = new HashSet<Link>());
              links.add(link);
            }
            case MODIFIED:
              conceptsToPersist.add(link.getFinalSourceConcept());
            break;
          }
        }
        else if (connectable instanceof Primitive) {
          try {
            PrimitiveImpl primitive = (PrimitiveImpl)connectable;
            if (primitive.getStatus() == Status.CREATED) {
              // Primitive's file copying is done here because after the internal commit the Primitive's status is lost.
              File src = new File((URI)primitive.getValue());
              File dst = new File(this.root, primitive.getLocalIdFragment().toString());
              if (!src.exists())
                  throw new TransactionException("Source for " + primitive.getId() + " does not exists:" + src.getAbsolutePath());
              try ( FileInputStream  is = new FileInputStream(src); FileOutputStream os = new FileOutputStream(dst)) {
                FileChannel srcfc = is.getChannel();
                FileChannel dstfc = os.getChannel();
                srcfc.transferTo(0, srcfc.size(), dstfc);
                // At last, we set Primitive's value to the copied file URI.
                this.setPrimitiveValue(primitive.getModificationTime(), primitive, dst.toURI());
              }
            }
          }
          catch (TransactionException ex) {
            throw ex;
          }
          catch (Exception ex) {
            throw new TransactionException(ex);
          }
        }
      }      
    }    
    // Then we commit transient internal structures.
    List<Id> result = super.commit(transactionId);
    // At last, we perform the persist modified elements.
    if (!(transactionId instanceof NotPersistentTransactionId)) {
      Exception error = null;
      for (Map.Entry<Id,Set<Link>> entry: linksToPersist.entrySet()) {
        try {
          persistLinksReferences(entry.getKey(), entry.getValue());
        }
        catch (Exception ex) {
          logger.error(ex.getMessage(), ex);
          error = ex;
        }
      }
      for (Concept concept: conceptsToPersist) {
        List<Link> toSave = new ArrayList<Link>(concept.getLinksCount());
        for (Link link: concept.getLinks().values()) {
          switch (((LinkImpl)link).getStatus()) {
            case UNTOUCHED:
              try {
                LinkImpl aux = ((LinkImpl)link).clone();
                this.cleanSource(aux); // We don't need to serialize the source.
                toSave.add(aux);
              }
              catch (CloneNotSupportedException ex) {
                logger.error(ex.getMessage(), ex);
              }
            case FIXED:
              break;
            default:
              logger.error("Link was left in wrong status after commit: " + link);
          }
        }
        try {
          try (ObjectOutputStream oos = new ObjectOutputStream(new FileOutputStream(fileForConceptLinks(concept.getId())))) {
            oos.writeObject(toSave.toArray(new LinkImpl[toSave.size()]));
          }
        }
        catch (Exception ex) {
          logger.error(ex.getMessage(), ex);
          error = ex;
        }
      }
      if (error != null)
        throw new TransactionException(error);
    }
    return result;
  }

  @Override
  public List<Id> rollback(TransactionId transactionId) throws TransactionException, MissingElementException {
    Set<ConnectableImpl> connectables = this.transactionPendingConnectables.get(transactionId);
    if (connectables != null) {
      for (ConnectableImpl connectable: connectables) {
        if ((connectable instanceof Primitive) && (connectable.getStatus() == Status.CREATED)) {
          File file = new File(this.root, ((Primitive)connectable).getLocalIdFragment().toString());
          if (file.length() == 0)
            file.delete();
          else
            throw new TransactionException("Transaction internal error in rollback. Trying to erase a file that should not be erased: " + file.getAbsolutePath());
        }
      }
    }
    return super.rollback(transactionId);
  }
}